package org.budo.warehouse.logic.consumer.mail;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.mail.HtmlEmail;
import org.budo.druid.util.DruidUtil;
import org.budo.graph.annotation.SpringGraph;
import org.budo.support.dao.page.Page;
import org.budo.support.freemarker.FreemarkerUtil;
import org.budo.support.javax.sql.util.JdbcUtil;
import org.budo.support.lang.util.MapUtil;
import org.budo.support.lang.util.StringUtil;
import org.budo.support.servlet.util.QueryStringUtil;
import org.budo.time.Time;
import org.budo.warehouse.logic.api.DataEntry;
import org.budo.warehouse.logic.api.DataMessage;
import org.budo.warehouse.logic.consumer.AbstractDataConsumer;
import org.budo.warehouse.logic.consumer.DataEntryPojo;
import org.budo.warehouse.logic.producer.DataMessageImpl;
import org.budo.warehouse.logic.util.PipelineUtil;
import org.budo.warehouse.service.api.IDataNodeService;
import org.budo.warehouse.service.api.IMailBufferService;
import org.budo.warehouse.service.entity.DataNode;
import org.budo.warehouse.service.entity.MailBuffer;
import org.budo.warehouse.service.entity.Pipeline;

import com.alibaba.fastjson.JSON;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
public class MailDataConsumer extends AbstractDataConsumer {
    private static final String BUFFER_SIZE_KEY = "bufferSize";

    private static final String MAX_DELAY_KEY = "maxDelay";

    private static final int MAX_DELAY_DEFAULT = 5;

    private static final int BUFFER_SIZE_DEFAULT = 2;

    private static final String MAIL_TEMPLATE = "<table border=\"1\">" //
            + "<tr>" //
            + "<td>ID</td>" //
            + "<td>Pipeline</td>" //
            + "<td>DataMessage</td>" //
            + "<td>CreatedAt</td>" //
            + "</tr>" //
            + "<#list mailBuffers as mailBuffer>" //
            + "<tr>" //
            + "<td>${mailBuffer.id?c}</td>" //
            + "<td>${mailBuffer.pipelineId?c}</td>" //
            + "<td>${mailBuffer.dataMessage}</td>" //
            + "<td>${mailBuffer.createdAt}</td>" //
            + "</tr>" //
            + "</#list>" + "</table>";

    @Resource
    private IMailBufferService mailBufferService;

    @Resource
    private IDataNodeService dataNodeService;

    @SpringGraph
    @Override
    public void consume(DataMessage dataMessage) {
        DataMessage dataMessagePojo = this.toPojo(dataMessage);

        if (dataMessagePojo.getDataEntries().isEmpty()) {
            log.warn("#44 dataEntryPojos=" + dataMessagePojo.getDataEntries() + ", dataMessage=" + dataMessage);
            return;
        }

        String dataMessageJson = JSON.toJSONString(dataMessagePojo);
        Integer pipelineId = this.getPipeline().getId();

        Integer targetDataNodeId = this.getPipeline().getTargetDataNodeId();
        DataNode targetDataNode = dataNodeService.findByIdCached(targetDataNodeId);

        // insert
        MailBuffer mailBuffer = new MailBuffer(pipelineId, dataMessageJson);
        mailBufferService.insert(mailBuffer);

        // 默认2条一批
        Integer bufferSize = QueryStringUtil.getParameterInteger(targetDataNode.getUrl(), BUFFER_SIZE_KEY, BUFFER_SIZE_DEFAULT);
        Integer count = mailBufferService.countNotSentByPipelineId(pipelineId);
        if (count >= bufferSize) {
            log.info("#72 bufferSize=" + bufferSize + ", count=" + count + ", pipeline=" + this.getPipeline() + ", targetDataNode=" + targetDataNode);
            this.sendMail(this.getPipeline()); // 发邮件
            return;
        }

        // 默认5秒延迟
        Integer maxDelay = QueryStringUtil.getParameterInteger(targetDataNode.getUrl(), MAX_DELAY_KEY, MAX_DELAY_DEFAULT);
        Timestamp maxCreatedAt = mailBufferService.findNotSentMaxCreatedAtByPipelineId(pipelineId);
        if (Time.now().isAfter(Time.when(maxCreatedAt).plusSecond(maxDelay))) {
            log.info("#81 maxDelay=" + maxDelay + ", maxCreatedAt=" + maxCreatedAt + ", pipeline=" + this.getPipeline() + ", targetDataNode=" + targetDataNode);
            this.sendMail(this.getPipeline()); // 发邮件
            return;
        }
    }

    private void sendMail(Pipeline pipeline) {
        try {
            DataNode targetDataNode = dataNodeService.findById(pipeline.getTargetDataNodeId());

            List<MailBuffer> mailBuffers = mailBufferService.listNotSentByPipelineId(pipeline.getId(), Page.max());
            log.info("#69 mailBuffers=" + mailBuffers + ", pipeline=" + pipeline);

            // mail:smtp://smtp.qq.com:465?bufferSize=5&maxDelay=5
            String url = targetDataNode.getUrl();
            String username = targetDataNode.getUsername();
            String password = targetDataNode.getPassword();
            password = DruidUtil.rsaDecrypt(password); // 密码解密

            String host = JdbcUtil.getHost(url);
            if (null == host || host.trim().isEmpty()) {
                log.error("#106 return, targetDataNode=" + targetDataNode);
                return;
            }

            String from = QueryStringUtil.getParameter(url, "from", username); // 默认用username
            String[] to = StringUtil.split(pipeline.getTargetTable(), ";");

            String subject = PipelineUtil.targetSchema(pipeline, null);

            // 邮件服务器配置
            HtmlEmail htmlEmail = new HtmlEmail();
            htmlEmail.setHostName(host);
            htmlEmail.setAuthentication(username, password);

            // 邮件正文
            Map<String, Object> map = MapUtil.stringObjectMap("mailBuffers", mailBuffers //
                    , "dataNodeService", this.dataNodeService);
            String mailContent = FreemarkerUtil.freemarkerRender(MAIL_TEMPLATE, map);

            // 发邮件
            String messageId = htmlEmail //
                    .addTo(to) //
                    .setFrom(from) //
                    .setSubject(subject)//
                    .setMsg(mailContent) //
                    .send();

            log.info("#124 from=" + from + ", to=" + Arrays.toString(to) + ", subject=" + subject + ", messageId=" + messageId);

            // 更新为已发送
            mailBufferService.updateSentAtByPipelineId(Time.now().toTimestamp(), pipeline.getId());
        } catch (Throwable e) {
            log.error("#111 pipeline=" + pipeline + ", e=" + e, e);
        }
    }

    private DataMessage toPojo(DataMessage dataMessage) {
        List<DataEntry> dataEntries = dataMessage.getDataEntries();
        List<DataEntry> dataEntryPojos = new ArrayList<DataEntry>();
        for (DataEntry dataEntry : dataEntries) {
            String schemaName = dataEntry.getSchemaName();
            String tableName = dataEntry.getTableName();
            String eventType = dataEntry.getEventType();

            if (tableName.equals("t_mail_buffer") && eventType.equals("INSERT")) {
                log.info("#135 dataEntry=" + dataEntry); // 避免造成递归
                continue;
            }

            // https://code.aliyun.com/wanshao/canal/commit/96dca0b9e6b6dcc12fc439d2f17d85a4656485df
            if (schemaName.equals("mysql") && tableName.equals("ha_health_check") && eventType.equals("CREATE")) {
                log.info("#140 dataEntry=" + dataEntry); // 避免造成递归
                continue;
            }

            dataEntryPojos.add(new DataEntryPojo(dataEntry));
        }

        return new DataMessageImpl(dataMessage.getDataNodeId(), dataEntryPojos);
    }
}